//====================================================================
// Filename:      goblin_sqs_queue_center.cc
//                
// It belongs to goblin-os.
//                
// Version:       
// Author:        Ins <InsFous@gmail.com>
// Created at:    Fri Dec 10 17:51:13 2010
// Description:   
//                
//====================================================================
#include "goblin_sqs_queue_center.h"

#include <string.h>

namespace goblin {
  namespace sqs {
    QueueCenter::QueueCenter(const std::string &db_dir) {
      db_dir_ = db_dir;
      pthread_rwlock_init(&rw_lock_, NULL);
      Init();
    }
    QueueCenter::~QueueCenter() {
      pthread_rwlock_wrlock(&rw_lock_);
      QueueStorage::iterator it = all_queues_.begin();
      for (; it != all_queues_.end(); ++it) {
	all_queues_.erase(it);
      }
      pthread_rwlock_unlock(&rw_lock_);
      pthread_rwlock_destroy(&rw_lock_);
    }
    void QueueCenter::Init() {
      char path[1024] = {0};
      snprintf(path, sizeof(path), "%s/qc", db_dir_.c_str());
      FILE *fp = fopen(path, "rb");
      if (fp != NULL) {
	std::string content;
	int n = -1;
	char buf[1024] = {0};
	while ((n = fread(buf, 1, sizeof(buf), fp)) > 0) {
	  content.append(buf, n);
	}
	const char *ptr = content.c_str();
	long left = content.length();
	while (left) {
	  struct QueueDisk *q = (struct QueueDisk *)ptr;
	  printf("Queue:{%s}\n", std::string(q->queue_name, q->queue_name_len).c_str());
	  struct QueueDesc q_desc = {new Queue(std::string(q->queue_name, q->queue_name_len), db_dir_), q->queue_flag};
	  all_queues_[std::string(q->queue_name, q->queue_name_len)] = q_desc;
	  ptr += sizeof(struct QueueDisk) + q->queue_name_len;
	  left -= sizeof(struct QueueDisk) + q->queue_name_len;
	}
	fclose(fp);
      }
    }
    bool QueueCenter::CreateQueue(const QueueName &queue_name, const QueueFlag::type queue_flag) {
      pthread_rwlock_wrlock(&rw_lock_);
      QueueStorage::iterator it = all_queues_.find(queue_name);
      if (it != all_queues_.end()) {
	const QueueFlag::type q_flag = (queue_flag == QueueFlag::None) ? QueueFlag::UsingExists : queue_flag ;
	if (q_flag & QueueFlag::UsingExists){
	  pthread_rwlock_unlock(&rw_lock_);
	  return true;
	} else if (q_flag & QueueFlag::ClearExists){
	  all_queues_.erase(it);
	  struct QueueDesc q_desc = {new Queue(queue_name, db_dir_), queue_flag};
	  all_queues_[queue_name] = q_desc; 
	  pthread_rwlock_unlock(&rw_lock_);
	  return true;
	}
	pthread_rwlock_unlock(&rw_lock_);
	return false;
      } else {
	struct QueueDesc q_desc = {new Queue(queue_name, db_dir_), queue_flag};
	all_queues_[queue_name] = q_desc;
      }
      pthread_rwlock_unlock(&rw_lock_);
      return true;
    }
    void QueueCenter::DeleteQueue(const QueueName &queue_name) {
      pthread_rwlock_wrlock(&rw_lock_);
      QueueStorage::iterator it = all_queues_.find(queue_name);
      if (it != all_queues_.end()) {
	all_queues_.erase(it);
      }
      pthread_rwlock_unlock(&rw_lock_);
    }
    bool QueueCenter::SendMessage(const QueueName &queue_name, const std::string &msg_content, const int32_t msg_type) {
      pthread_rwlock_rdlock(&rw_lock_);
      QueueStorage::const_iterator cit = all_queues_.find(queue_name);
      if (cit != all_queues_.end()) {
	cit->second.queue->PushMessage(msg_content, msg_type);
	pthread_rwlock_unlock(&rw_lock_);
	return true;
      }
      pthread_rwlock_unlock(&rw_lock_);
      return false;
    }
    
    bool QueueCenter::FetchMessage(const QueueName &queue_name, Message &msg) {
      pthread_rwlock_rdlock(&rw_lock_);
      QueueStorage::const_iterator cit = all_queues_.find(queue_name);
      if (cit != all_queues_.end()) {
	bool b = cit->second.queue->GetMessage(msg);
	pthread_rwlock_unlock(&rw_lock_);
	return b;
      }
      pthread_rwlock_unlock(&rw_lock_);
      return false;
    }

    int QueueCenter::FetchMessageList(const QueueName &queue_name, int count, std::vector<Message> &msg_list) {
      pthread_rwlock_rdlock(&rw_lock_);
      QueueStorage::const_iterator cit = all_queues_.find(queue_name);
      if (cit != all_queues_.end()) {
	int fetched_count = 0;
	Message msg;
	bool ret = false;
	while ((ret = cit->second.queue->GetMessage(msg))) {
	  msg_list.push_back(msg);
	  ++fetched_count;
	}
	pthread_rwlock_unlock(&rw_lock_);
	return fetched_count;
      }
      pthread_rwlock_unlock(&rw_lock_);
      return false;
    }
    
    void QueueCenter::GetQueueStatus(QueueStatus &_return, const QueueName &queue_name) {
      pthread_rwlock_rdlock(&rw_lock_);
      QueueStorage::const_iterator cit = all_queues_.find(queue_name);
      if (cit != all_queues_.end()) {
	_return.queue_name = queue_name;
	_return.capacity = cit->second.queue->Capacity();
	_return.unread_msg_count  = cit->second.queue->UnreadMsgCount();
	_return.total_msg_count = cit->second.queue->TotalMsgCount();
      }
      pthread_rwlock_unlock(&rw_lock_);
    }
    
    bool QueueCenter::SetQueueCapacity(const QueueName &queue_name, const int64_t capacity) {
      pthread_rwlock_rdlock(&rw_lock_);
      QueueStorage::const_iterator cit = all_queues_.find(queue_name);
      if (cit != all_queues_.end()) {
	bool b = cit->second.queue->SetCapacity(capacity);
	pthread_rwlock_unlock(&rw_lock_);
	return b;
      }
      pthread_rwlock_unlock(&rw_lock_);
      return false;
    }

    void QueueCenter::SetQueueFilter(const QueueName &queue_name, const int32_t msg_type) {
      pthread_rwlock_rdlock(&rw_lock_);
      QueueStorage::const_iterator cit = all_queues_.find(queue_name);
      if (cit != all_queues_.end()) {
	cit->second.queue->SetFilter(msg_type);
      }
      pthread_rwlock_unlock(&rw_lock_);
    }

    void QueueCenter::RestoreQueueDefaultFilter(const QueueName &queue_name) {
      pthread_rwlock_rdlock(&rw_lock_);
      QueueStorage::const_iterator cit = all_queues_.find(queue_name);
      if (cit != all_queues_.end()) {
	cit->second.queue->RestoreDefaultFilter();
      }
      pthread_rwlock_unlock(&rw_lock_);
    }

    void QueueCenter::SyncToDisk() {
      pthread_rwlock_rdlock(&rw_lock_);
      QueueStorage::const_iterator cit = all_queues_.begin();
      char path[1024] = {0};
      snprintf(path, sizeof(path), "%s/qc", db_dir_.c_str());
      FILE *fp = fopen(path, "wb+");
      for (; cit != all_queues_.end(); ++cit) {
	struct QueueDisk *q = (struct QueueDisk *)malloc(sizeof(struct QueueDisk) + cit->first.length());
	q->queue_flag = cit->second.queue_flag;
	q->queue_name_len = cit->first.length();
	memcpy(q->queue_name, (char *)cit->first.c_str(), cit->first.length());
	fwrite((char *)q, 1, sizeof(struct QueueDisk) + cit->first.length(), fp);
	cit->second.queue->SyncToDisk();
      }
      pthread_rwlock_unlock(&rw_lock_);
      fclose(fp);
    }

    void QueueCenter::Save() {
      SyncToDisk();
      pthread_rwlock_rdlock(&rw_lock_);
      QueueStorage::const_iterator cit = all_queues_.begin();
      for (; cit != all_queues_.end(); ++cit) {
	cit->second.queue->Save();
      }
      pthread_rwlock_unlock(&rw_lock_);
    }

  }
}
